package ext

import (
	"crypto/tls"
	"fmt"
	"github.com/segmentio/kafka-go/sasl"
	"strings"
	"time"

	"github.com/icetech233/ekuiper-kafka/conf"
	kafkago "github.com/segmentio/kafka-go"
)

type Closable interface {
	Close(ctx StreamContext) error
}

type Source interface {
	// Open Should be sync function for normal case. The container will run it in go func
	Open(ctx StreamContext, consumer chan<- SourceTuple, errCh chan<- error)
	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
	// read from the yaml
	Configure(datasource string, props map[string]interface{}) error
	Closable
}

type KafkaSource struct {
	reader    *kafkago.Reader
	offset    int64
	tlsConfig *tls.Config
	sc        *kafkaSourceConf
}

func (s *KafkaSource) Ping(d string, props map[string]interface{}) error {
	err := s.Configure(d, props)
	if err != nil {
		return err
	}
	for _, broker := range strings.Split(s.sc.Brokers, ",") {
		err = ping(s.tlsConfig, broker)
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *KafkaSource) Configure(topic string, props map[string]interface{}) error {
	if len(topic) < 1 {
		conf.Log.Error("DataSource which indicates the topic should be defined")
		return fmt.Errorf("DataSource which indicates the topic should be defined")
	}
	var err error
	var tlsConfig *tls.Config
	var kConf *kafkaSourceConf
	var saslConf SaslConf
	var mechanism sasl.Mechanism
	var readerConfig kafkago.ReaderConfig

	kConf, err = getSourceConf(props)
	if err != nil {
		conf.Log.Errorf("kafka source config error: %v", err)
		return err
	}
	if err = kConf.validate(); err != nil {
		return err
	}
	tlsConfig, err = conf.GenTLSConfig(props, "kafka-source")
	if err != nil {
		conf.Log.Errorf("kafka tls conf error: %v", err)
		return err
	}
	saslConf, err = GetSaslConf(props)
	if err != nil {
		conf.Log.Errorf("kafka sasl error: %v", err)
		return err
	}
	if err = saslConf.Validate(); err != nil {
		conf.Log.Errorf("kafka validate sasl error: %v", err)
		return err
	}
	mechanism, err = saslConf.GetMechanism()
	if err != nil {
		conf.Log.Errorf("kafka sasl mechanism error: %v", err)
		return err
	}
	readerConfig = kConf.GetReaderConfig(topic)
	conf.Log.Infof("topic: %s, brokers: %v", readerConfig.Topic, readerConfig.Brokers)
	readerConfig.Dialer = &kafkago.Dialer{
		Timeout:       10 * time.Second,
		DualStack:     true,
		TLS:           tlsConfig,
		SASLMechanism: mechanism,
	}
	reader := kafkago.NewReader(readerConfig)
	s.reader = reader
	s.sc = kConf
	if err = s.reader.SetOffset(kafkago.LastOffset); err != nil {
		return err
	}
	conf.Log.Infof("kafka source got configured.")
	return nil
}

func (s *KafkaSource) Close(_ StreamContext) error {
	return nil
}

func ping(tlsConfig *tls.Config, address string) error {
	d := &kafkago.Dialer{TLS: tlsConfig}
	c, err := d.Dial("tcp", address)
	if err != nil {
		return err
	}
	defer c.Close()
	return nil
}
